package com.bwie.controller;

import com.bwie.config.KafkaConfig;
import com.bwie.utils.R;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author 军哥
 * @version 1.0
 * @description: TODO
 * @date 2025/3/27 15:22
 */

@RestController
@RequestMapping("/user/kafka")
public class KafkaController {

    @Autowired
    KafkaTemplate kafkaTemplate;

    /**
     * @description 发送Kafka消息
     * @params [msg]
     * @return com.bwie.utils.R
     * @author 军哥
     * @date 2025/3/27 15:28
     */
    @PostMapping("/send/{msg}")
    public R send(@PathVariable("msg") String msg){
        System.out.println("发送消息："+msg);
        kafkaTemplate.send(KafkaConfig.TOPIC_TEST, msg);
        return R.OK(msg);
    }


    /**
     * @description 接收Kafka消息
     * @params [msg, acknowledgment]
     * @return void
     * @author 军哥
     * @date 2025/3/27 15:30
     */
    @KafkaListener(topics = KafkaConfig.TOPIC_TEST)
    public void recvTestMessage(String msg, Acknowledgment acknowledgment){
        System.out.println("接收到消息："+msg);

        // 消息手动确认：删除消息
        acknowledgment.acknowledge();
    }

}
